package com.itheima.kafka.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.ExecutionException;

/**
 * Kafka生产者启动类
 */
@SpringBootApplication
public class KafkaProducerApplication {
    
    // 定义日志记录器
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducerApplication.class);
    
    // 定义发送消息的模板类
    private static KafkaTemplate kafkaTemplate;
    
    public static void main(String[] args) {
        
        // 初始化环境
        ConfigurableApplicationContext applicationContext = SpringApplication.run(KafkaProducerApplication.class, args);
        kafkaTemplate = applicationContext.getBean(KafkaTemplate.class);
        
        // 测试Kafka环境
        // environmentTest();
        
        //发后即忘
        // sendMessageMethod01();
        
        //同步消息发送
        // sendMessageMethod02();
        
        //异步消息发送
        // sendMessageMethod03();
        
        //发送消息到指定分区,才还能保证分区消息的顺序性
        // sendMessage2DestPartition();
        
        //事务消息的第一种方式
        // sendTransactionMessageMethod01();
        
        //事务消息的第二种方式 
        sendTransactionMessage(applicationContext);
        
    }
    
    private static void sendTransactionMessage(ConfigurableApplicationContext applicationContext) {
        ProducerMq producerMq = applicationContext.getBean(ProducerMq.class);
        producerMq.sendTransactionMessage();
    }
    
    private static void sendTransactionMessageMethod01() {
        kafkaTemplate.executeInTransaction(operations -> {
            //发送消息
            kafkaTemplate.send("itcast", 0, "order", "事务消息 ------>1");
            //发送消息
            kafkaTemplate.send("itcast", 0, "order", "事务消息 ------>2");
            // int i = 1 / 0;
            //发送消息
            kafkaTemplate.send("itcast", 0, "order", "事务消息 ------>3");
            return true;
        });
        //发送到的是recordAccumulator
        LOGGER.info("send transaction message to local cache success");
    }
    // private static void sendTransactionMessageMethod01() {
    //     kafkaTemplate.executeInTransaction(operations -> {
    //         // 发送消息
    //         kafkaTemplate.send("itcast", 0, "order", "事务消息 ----> 1")
    //         ;
    //         kafkaTemplate.send("itcast", 0, "order", "事务消息 ----> 2")
    //         ;
    //         // 产生异常代码
    //         // int a = 1 / 0;
    //         kafkaTemplate.send("itcast", 0, "order", "事务消息 ----> 3")
    //         ;
    //         return true;
    //     });
    //     LOGGER.info("send transaction message to local cache success ");
    // }
    
    public static void sendMessage2DestPartition() {
        for (int i = 0; i < 5; i++) {
            String msg = "kafka 环境测试" + i;
            ListenableFutureCallback listenableFutureCallback = new ListenableFutureCallback() {
                @Override
                public void onFailure(Throwable throwable) {
                    LOGGER.info("发生了异常,异常为:", throwable);
                }
                
                @Override
                public void onSuccess(Object obj) {
                    LOGGER.info("发送成功!  {}", ((SendResult) obj).getProducerRecord().value());
                }
            };
            kafkaTemplate.send("itcast", 0, "order", msg).addCallback(listenableFutureCallback);
            LOGGER.info("发送数据到本地缓存成功---->{}", msg);
        }
        
    }
    
    // private static void sendMessage2DestPartition() {
    //     for (int i = 0; i < 5; i++) {
    //         String msg = "kafka 环境测试" + i;
    //        
    //         ListenableFutureCallback listenableFutureCallback = new ListenableFutureCallback() {
    //             @Override
    //             public void onFailure(Throwable throwable) {
    //                
    //                 throwable.printStackTrace();
    //             }
    //            
    //             @Override
    //             public void onSuccess(Object o) {
    //                 LOGGER.info("发送成功--->{}", ((SendResult) o).getProducerRecord().value());
    //             }
    //         };
    //         kafkaTemplate.send("itcast", 0, "order", msg)
    //                 .addCallback(listenableFutureCallback);
    //         LOGGER.info("send msg to local cache success ----->", msg);
    //     }
    // }
    
    // Kafka单节点环境测试
    public static void environmentTest() {
        kafkaTemplate.send("itheima", "Kakfa环境测试....");
    }
    
    // 演示消息发送： 发送即忘,这个性能好,可靠性差
// 演示消息发送： 发送即忘
    public static void sendMessageMethod01() {
        for (int x = 0; x < 5; x++) {
            String msg = "Kakfa环境测试...." + x;
            kafkaTemplate.send("itheima", msg); // 发后即忘
            LOGGER.info("send msg success ---> {} ", msg);
        }
    }
    
    // 演示消息发送：同步发送,这个阻塞kafka响应,性能差,可靠性好
    
    private static void sendMessageMethod02() {
        for (int i = 0; i < 5; i++) {
            
            try {// 同步消息发送
                kafkaTemplate.send("itheima", "发送的消息" + i).get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }
    
    // 演示消息发送： 异步发送,性能好,可靠性有保障
    private static void sendMessageMethod03() {
        for (int i = 0; i < 5; i++) {
            
            ListenableFutureCallback listenableFutureCallback = new ListenableFutureCallback() {
                @Override
                public void onFailure(Throwable throwable) {
                    
                    throwable.printStackTrace();
                }
                
                @Override
                public void onSuccess(Object o) {
                    LOGGER.info("发送成功--->{}", ((SendResult) o).getProducerRecord().value());
                }
            };
            kafkaTemplate.send("itheima", "发送的消息" + i)
                    .addCallback(listenableFutureCallback);
        }
    }
    
}
